package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: snn
 * @Description: 针对企查查t_eci_partner表数据映射开发
 * @Date: 2021/12/13 17:51
 */

public class TEciPartnerZs {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("zs_t_eci_partner").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_stockholder limit 10");
        long zs_count = zs_data.count();
        if(zs_count>0){
//            spark.sql("select * from a_dc_ep_ods.t_eci_company where instr(status,'销')=0").createOrReplaceTempView("qcc_t_eci_company");
            spark.sql("select * from a_dc_ep_ods.t_eci_company").createOrReplaceTempView("qcc_t_eci_company");
            spark.sql("select * from a_dc_ep_ods.com_company").createOrReplaceTempView("zs_com_company");
            //dwi中的t_eci_partner表数据关联获取credit_code
            spark.sql("select a.*,b.credit_code from a_dc_ep_dwi.zs_t_eci_partner a left join qcc_t_eci_company b on a.company_id = b.company_id").createOrReplaceTempView("qcc_t_eci_partner");
            //获取zs_t_stockholder_del表中所有creditcode不为空的数据
//            spark.sql("select b.* from qcc_t_eci_partner b inner join (select a.* from a_dc_ep_ods.zs_t_stockholder_del a where a.creditcode is not null and a.creditcode != '') c on b.credit_code = c.creditcode");
//            spark.sql("select b.* from qcc_t_eci_partner b inner join (select d.name from (select a.* from a_dc_ep_ods.zs_t_stockholder_del a where a.creditcode is null or a.creditcode = '') e inner join zs_com_company d on e.entid = d.entid where d.name is not null and d.name != '') c on c.name = b.company_name");

            spark.sql("select tmp.* from (select t.*,if(s.name is not null,s.name,qcc.company_name) as entname from a_dc_ep_ods.zs_t_stockholder_del t left join  zs_com_company s on t.entid = s.entid left join (select * from qcc_t_eci_company where credit_code is not null and credit_code != '') qcc on qcc.credit_code=t.creditcode) tmp where tmp.entname is not null").createOrReplaceTempView("zs_t_stockholder_del_1");
            //获取qcc_t_eci_partner表中所有需要删除的数据即zs_t_stockholder_del表中所有数据
            Dataset<Row> datacCedit_code = spark.sql("select distinct a.* from (select * from qcc_t_eci_partner where credit_code in (select creditcode from zs_t_stockholder_del_1 where length(creditcode)>=10) union all select * from qcc_t_eci_partner where company_name in (select entname from zs_t_stockholder_del_1 where length(creditcode)<10 and length(trim(entname))!=0)) as a");
            if(datacCedit_code.count()>0){
                datacCedit_code.createOrReplaceTempView("qcc_t_eci_partner_del");
            }else{
                //获取zs_t_stockholder_del表中所有creditcode为空的数据吗，并关联zs_t_stockholder表获取企业名称
                spark.sql("select distinct a.entname from zs_t_stockholder_del_1 a left join a_dc_ep_ods.zs_t_stockholder b on a.entid = b.entid where (a.creditcode is null or a.creditcode = '') and a.entname is not null").createOrReplaceTempView("zs_t_stockholder_del_2");
                spark.sql("select * from qcc_t_eci_partner where company_name in (select entname from zs_t_stockholder_del_2)").createOrReplaceTempView("qcc_t_eci_partner_del");
            }
            //将删除表中的数据插入到zs_t_eci_partner表中
            spark.sql("insert overwrite table a_dc_ep_incr.zs_t_eci_partner select id,key_no,company_id,company_name,stock_name,stock_type,stock_percent,should_capi,should_capi_value,should_capi_unit,shoud_date,invest_type,invest_name,real_capi,real_capi_value, real_capi_unit,capi_date,p_key_no,dates,'-1' as isadd from qcc_t_eci_partner_del");
            //将zs_t_stockholder表和qcc_t_eci_partner表中的数据进行关联，将zs_t_stockholder表中的数据插入到zs_t_eci_partner表中
            spark.sql("select \n" +
                    "md5(concat(b.company_name,a.shaname)) as id,\n" +
                    "if(length(b.company_id)>20,b.company_id,md5(if(length(a.creditcodes)=18,a.creditcodes,regexp_replace(trim(b.company_name),'([(（）)])','')))) as key_no,\n" +
                    "if(length(b.company_id)>20,b.company_id,md5(if(length(a.creditcodes)=18,a.creditcodes,regexp_replace(trim(b.company_name),'([(（）)])','')))) as company_id,\n" +
                    "b.company_name as company_name,\n" +
                    "a.shaname as stock_name,\n" +
                    "a.invtype as stock_type,\n" +
                    "a.fundedratio  as stock_percent,\n" +
                    "a.subconam  as should_capi,\n" +
                    "a.subconam  as should_capi_value,\n" +
                    "a.regcapcur  as should_capi_unit,\n" +
                    "a.condate  as shoud_date,\n" +
                    "a.conform  as invest_type,\n" +
                    "a.conform  as invest_name,\n" +
                    "a.acconam  as real_capi,\n" +
                    "a.acconam  as real_capi_value,\n" +
                    "a.regcapcur as real_capi_unit,\n" +
                    "a.condate as capi_date,\n" +
                    //invtype!='20' 是股东类型代码20，代表自然人股东
                    //"if(length(c.company_id)>20,c.company_id,if(invtype!='20',md5(if(length(a.creditcode)=18,a.creditcode,s.name)),md5(concat('zjs_',a.shaname)))) as p_key_no,\n" +
                    "md5(concat(b.company_name,a.shaname)) as p_key_no,\n" +
                    "a.jobid as dates,\n" +
                    "'1' as isadd, \n" +
                    "a.invtypecode, "+
                    "a.creditcode " +
                    "from (select k.*,j.creditcode as creditcodes from a_dc_ep_ods.zs_t_stockholder k left join zs_t_stockholder_del_1 j on k.entid=j.entid where j.creditcode is not null and j.creditcode !='') a\n" +
//                            "inner join zs_com_company s on a.entid = s.entid \n" +
                    "inner join qcc_t_eci_company b on a.creditcodes = b.credit_code")
                    .union(spark.sql("select \n" +
                            "md5(concat(b.company_name,a.shaname)) as id,\n" +
                            "if(length(b.company_id)>20,b.company_id,md5(if(length(a.creditcodes)=18,a.creditcodes,regexp_replace(trim(b.company_name),'([(（）)])','')))) as key_no,\n" +
                            "if(length(b.company_id)>20,b.company_id,md5(if(length(a.creditcodes)=18,a.creditcodes,regexp_replace(trim(b.company_name),'([(（）)])','')))) as company_id,\n" +
                            "b.company_name as company_name,\n" +
                            "a.shaname as stock_name,\n" +
                            "a.invtype as stock_type,\n" +
                            "a.fundedratio  as stock_percent,\n" +
                            "a.subconam  as should_capi,\n" +
                            "a.subconam  as should_capi_value,\n" +
                            "a.regcapcur  as should_capi_unit,\n" +
                            "a.condate  as shoud_date,\n" +
                            "a.conform  as invest_type,\n" +
                            "a.conform  as invest_name,\n" +
                            "a.acconam  as real_capi,\n" +
                            "a.acconam  as real_capi_value,\n" +
                            "a.regcapcur as real_capi_unit,\n" +
                            "a.condate as capi_date,\n" +
                            //invtype!='20' 是股东类型代码20，代表自然人股东
                            //"if(length(c.company_id)>20,c.company_id,if(invtype!='20',md5(if(length(a.creditcode)=18,a.creditcode,s.name)),md5(concat('zjs_',a.shaname)))) as p_key_no,\n" +
                            "md5(concat(b.company_name,a.shaname)) as p_key_no,\n" +
                            "a.jobid as dates,\n" +
                            "'1' as isadd, \n" +
                            "a.invtypecode, "+
                            "a.creditcode " +
                            "from (select k.*,j.creditcode as creditcodes,j.entname as name from a_dc_ep_ods.zs_t_stockholder k left join zs_t_stockholder_del_1 j on k.entid=j.entid where j.creditcode is null or j.creditcode = '') a\n" +
//                            "inner join zs_com_company s on a.entid = s.entid \n" +
                            "inner join qcc_t_eci_company b on regexp_replace(trim(a.name),'([(（）)])','') = regexp_replace(trim(b.company_name),'([(（）)])','')")).createOrReplaceTempView("zs_t_stockholder_1");

            spark.sql("select b.*,if(length(c.company_id)>20,c.company_id,md5(b.creditcode)) as stock_company_id from (select a.* from zs_t_stockholder_1 a where a.creditcode is not null and a.creditcode != '') b left join qcc_t_eci_company c on b.creditcode = c.credit_code")
                    .union(spark.sql("select b.*,if(length(c.company_id)>20,c.company_id,md5(c.company_name)) as stock_company_id from (select a.* from zs_t_stockholder_1 a where (a.creditcode is null or a.creditcode = '') and a.stock_name is not null and a.stock_name != '') b left join qcc_t_eci_company c on b.stock_name = c.company_name"))
                    .createOrReplaceTempView("zs_t_stockholder_2");
            //将zs_t_stockholder_1表中的数据插入到zs_t_eci_partner表中
            spark.sql("insert into table a_dc_ep_incr.zs_t_eci_partner select a.id,a.key_no,a.company_id,a.company_name,a.stock_name,a.stock_type,a.stock_percent,a.should_capi,a.should_capi_value,a.should_capi_unit,a.shoud_date,a.invest_type,a.invest_name,a.real_capi,a.real_capi_value,a.real_capi_unit,a.capi_date,if(a.invtypecode != '20',a.stock_company_id,a.p_key_no) as p_key_no,a.dates,a.isadd  from (select *,ROW_NUMBER() OVER(PARTITION BY concat(company_id,p_key_no) ORDER BY dates DESC,isadd DESC) num from  zs_t_stockholder_2) a where a.num=1 ");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.zs_t_eci_partner");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
